package com.we.risk.antirefund.gps;

import com.alibaba.fastjson.JSONObject;
import com.we.flink.utils.WeKafkaPartitioner;
import com.we.flink.utils.WeKafkaPropertyReader;
import com.we.risk.antirefund.phoneinfo.HfqPhoneInfo;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.sink.TopicSelector;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.util.Iterator;
import java.util.Properties;

public class AdmGpsRecordLabel {
    public static final String RELEASEPROP = "risk/antirefund/gps/adm_gps_record_kfk_prod.properties";
    public static Logger LOG = LoggerFactory.getLogger(AdmGpsRecordLabel.class);

    public static final String KFKBROKERHOSTBAK =
            "10.10.13.22:9092,10.10.13.23:9092,10.10.13.24:9092";
    public static final String KFKBAKTOPIC = "kafka.topic.data.record";
    public static final String KFKBAKTOPICGRPID = "Gps_Record_grp_s2";

    public static final String MSGTYPE = "HFQ_UPLOAD_GPS_RECORD";
    public static final String INPUTKEYBY = "userKey";
    public static final String OUTPUTKEY = "user_key";
    public static final String HTTPTIMESTAMP = "httpTimestamp";
    public static final String CITY = "city";
    public static final String CAPTIME = "capture_time";
    public static final long RECVSTARTTIME = 1679932800L; //2023-03-28 00:00:00

    public static final String SINKTOPIC = "adm_anti_refund_gps_record";


    public static void main(String[] args) throws IOException {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        try {
            WeKafkaPropertyReader paramReader = WeKafkaPropertyReader.init(RELEASEPROP);
            /** RocksDB */
            env.setStateBackend(new RocksDBStateBackend(paramReader.getRocksDBBackendUrl()));
            /** checkpoint configure */
            CheckpointConfig ckConf = env.getCheckpointConfig();
            ckConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            ckConf.setCheckpointInterval(10 * 60 * 1000); // ms
            ckConf.setCheckpointTimeout(60 * 60 * 1000);
            ckConf.setMaxConcurrentCheckpoints(1);
            ckConf.setMinPauseBetweenCheckpoints(500);
            ckConf.enableExternalizedCheckpoints(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

            ckConf.enableUnalignedCheckpoints();

            /** Consumer Kafka */
            String srcKafkaTopic = paramReader.getTmpKfkTopic();
            String srcKafkaBootStrapServer = paramReader.getTmpKfkBootStrapServer();
            String srckfkGrupId = paramReader.getTmpKfkGroupId();
            /** sink to Kafka */
            String sinkKfkAddress = paramReader.getKfkBootStrapServer();
            int sinkkfkPartitions = paramReader.getKfkPartitions();

            Properties consumProp = new Properties();
            consumProp.setProperty("group.id", srckfkGrupId);
            consumProp.setProperty(
                    "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            consumProp.setProperty(
                    "value.deserializer",
                    "org.apache.kafka.common.serialization.StringDeserializer");

            /** set offset to latest */
            KafkaSource<String> kfkSource =
                    KafkaSource.<String>builder()
                            .setBootstrapServers(srcKafkaBootStrapServer)
                            .setTopics(srcKafkaTopic)
                            .setStartingOffsets(OffsetsInitializer.earliest())
                            .setProperties(consumProp)
                            .setValueOnlyDeserializer(new SimpleStringSchema())
                            .build();

            consumProp.setProperty("group.id", KFKBAKTOPICGRPID);
            KafkaSource<String> kfkBakSource =
                    KafkaSource.<String>builder()
                            .setBootstrapServers(KFKBROKERHOSTBAK)
                            .setTopics(KFKBAKTOPIC)
                            .setStartingOffsets(OffsetsInitializer.latest())
                            .setProperties(consumProp)
                            .setValueOnlyDeserializer(new SimpleStringSchema())
                            .build();

            SingleOutputStreamOperator<String> input =
                    env.fromSource(kfkSource, WatermarkStrategy.noWatermarks(), "gpsRecord")
                            .uid("kfk-gpsRecord-source");
            SingleOutputStreamOperator<String> inputBak =
                    env.fromSource(kfkBakSource, WatermarkStrategy.noWatermarks(), "gpsRecord-bak")
                            .uid("kfk-gpsRecord-bak-source");
            DataStream<String> unionInput = input.union(inputBak);

            SingleOutputStreamOperator<String> out = unionInput.flatMap(new gpsRecordRichFlatMapFunction());

            KafkaRecordSerializationSchema<String> serializedRecord =
                    KafkaRecordSerializationSchema.builder()
                            .setTopicSelector(
                                    new phoneInfoTopicSelector())
                            .setKeySerializationSchema(
                                    new SerializationSchema<String>() {
                                        @Override
                                        public byte[] serialize(String element) {
                                            JSONObject input = JSONObject.parseObject(element);
                                            String key = input.getString(OUTPUTKEY);
                                            return key.getBytes();
                                        }
                                    })
                            .setValueSerializationSchema(
                                    new SerializationSchema<String>() {
                                        @Override
                                        public byte[] serialize(String element) {
                                            return element.getBytes();
                                        }
                                    })
                            .setPartitioner(new WeKafkaPartitioner())
                            .build();

            KafkaSink<String> kfkSink =
                    KafkaSink.<String>builder()
                            .setBootstrapServers(sinkKfkAddress)
                            .setRecordSerializer(serializedRecord)
                            .build();

            out.sinkTo(kfkSink).setParallelism(sinkkfkPartitions);

            env.execute(AdmGpsRecordLabel.class.toString());
        } catch (Exception e) {
            LOG.error("Exception: " + e.getMessage());
            e.printStackTrace();
        }
    }

    private static class phoneInfoTopicSelector implements TopicSelector<String> {
        @Override
        public String apply(String s) {
            return SINKTOPIC;
        }
    }


    private static class gpsRecordRichFlatMapFunction extends RichFlatMapFunction<String, String> {
        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {
            try {
                GpsRecord gpsRecord = filterGpsRecordMsg(value);
                LOG.debug("hfqPhoneInfo: " + gpsRecord.toString());
                String user_key = gpsRecord.getUser_key();
                long cap_time = gpsRecord.getCap_time();
                if(cap_time != 0 && cap_time >= RECVSTARTTIME) {
                    String city = gpsRecord.getCity();
                    if (user_key != null && city != null && !"".equals(city)) {
                        JSONObject outJson = new JSONObject();
                        outJson.put(OUTPUTKEY, user_key);
                        outJson.put(CAPTIME, cap_time);
                        outJson.put(CITY, city);
                        LOG.debug("deviceID outJson: " + outJson.toString());
                        out.collect(outJson.toString());
                    } else {
                        LOG.warn("gpsRecordRichFlatMapFunction UserKey or Captime or City error!!");
                    }
                }
            } catch (Exception e) {
                LOG.error(e.getMessage());
                LOG.error("phoneInfoRichFlatMapFunction Exception input: " + value);
            }
        }

    }

    private static GpsRecord filterGpsRecordMsg(String value) {
        GpsRecord gpsRecord = new GpsRecord();
        try {
            if (value != null && value.contains(MSGTYPE)) {
                JSONObject inputJson = JSONObject.parseObject(value);
                if (inputJson != null) {
                    String type = inputJson.getString("type");
                    if (type != null && type.equals(MSGTYPE)) {
                        JSONObject datajson = inputJson.getJSONObject("dataJson");
                        if (datajson != null) {
                            String userKey = datajson.getString(INPUTKEYBY);
                            gpsRecord.setUser_key(userKey);

                            JSONObject data = datajson.getJSONObject("data");
                            if (data != null) {
                                String city = data.getString(CITY);
                                gpsRecord.setCity(city);

                                JSONObject baseData = data.getJSONObject("baseData");
                                if (baseData != null) {
                                    long capture_time = baseData.getLongValue(HTTPTIMESTAMP);
                                    gpsRecord.setCap_time(capture_time);
                                }
                            }
                        }
                    }
                }
            }
        } catch (Exception e) {
            LOG.error(e.getMessage());
            LOG.error("filterSmsInfoMsg excep input: " + value);
        }
        return gpsRecord;
    }


    private static JSONObject initPropJson(String propFile) throws IOException {
        InputStream resourceAsStream =
                AdmGpsRecordLabel.class.getClassLoader().getResourceAsStream(propFile);
        ParameterTool reader = ParameterTool.fromPropertiesFile(resourceAsStream);
        JSONObject res = new JSONObject();
        Iterator<Object> keysIterator = reader.getProperties().keySet().iterator();
        while (keysIterator.hasNext()) {
            String key = (String) keysIterator.next();
            if (key.equals(CAPTIME)) {
                res.put(key, -1);
            } else {
                res.put(key, "initV");
            }
        }
        return res;
    }

    private static class WeKafkaKeyedSerializationSchema
            implements KeyedSerializationSchema<String> {
        @Override
        public byte[] serializeKey(String element) {
            JSONObject jsonObject = JSONObject.parseObject(element);
            String keyby = jsonObject.getString(OUTPUTKEY);
            return keyby.getBytes();
        }

        @Override
        public byte[] serializeValue(String element) {
            return element.getBytes();
        }

        @Override
        public String getTargetTopic(String element) {
            return null;
        }
    }

    private static class WeKafkaCustomPartitioner extends FlinkKafkaPartitioner<String> {

        @Override
        public int partition(
                String record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
            int partition = Math.abs(new String(key).hashCode() % partitions.length);
            if (LOG.isDebugEnabled()) {
                LOG.debug(
                        " partitions: "
                                + partitions.length
                                + " partition: "
                                + partition
                                + " key: "
                                + new String(key));
            }
            return partition;
        }
    }
}
